#include "sys.h"
#include "log.h"
#include "dial.h"
#include "cJSON.h"
#include "dns_tactics.h"
#include "kafka.h"

void* do_dns_tactics(void* arg)
{
    dns_tactics_t* dns = (dns_tactics_t*)arg;
    dnsTactics_t* ptr;
    struct dnsTactics_head* head = &dns->head;
    cJSON *result, *obj, *arr;

    dial_arg_t dial_arg = { 0 };
    dial_result_t* dial_result = NULL;
    dns_result_t* dns_result = NULL;
    dns_result_t* tcpdns_result = NULL;
    char tmp_str[32] = { 0 };
    char topic[32] = { 0 };
    char *post_data = NULL, reportTime[32], collectTime[32];
    
    if (dns->count == 0)
        return NULL;
    result = cJSON_CreateObject();
    cJSON_AddStringToObject(result, "dialTaskId", dns->taskId);
    cJSON_AddStringToObject(result, "version", dns->version);
    cJSON_AddStringToObject(result, "indexType", "dns");
    get_timestring(reportTime);
    cJSON_AddStringToObject(result, "reportTime", reportTime);
    get_timestring(collectTime);
    cJSON_AddStringToObject(result, "collectTime", collectTime);
    // pthread_mutex_lock(&dns->mutex);
    arr = cJSON_CreateArray();
    TAILQ_FOREACH(ptr, head, entry)
    {
        obj = cJSON_CreateObject();
        if (!strcmp(ptr->type, "tcp")) {//tcp dns
            cJSON_AddNumberToObject(obj, "errno", 0);
            cJSON_AddStringToObject(obj, "dnsAdr", "0.0.0.0");
            cJSON_AddStringToObject(obj, "delay", "0.00");
        } else { 
            strcpy(dial_arg.in.dns_arg.dns_svr, ptr->ip);
            dial_arg.in.dns_arg.port = htons(ptr->port);
            strcpy(dial_arg.in.dns_arg.domain, ptr->domainName);
            dial_result = (dial_result_t*)dial_dns(&dial_arg);
            dns_result_print(&dial_result->out.dns_result);

            dns_result = &dial_result->out.dns_result;
            cJSON_AddNumberToObject(obj, "errno", dns_result->rcode);
            strcpy(tmp_str, inet_ntoa(*((struct in_addr*)&dns_result->ipaddr[0])));
            cJSON_AddStringToObject(obj, "dnsAdr", tmp_str);
            sprintf(tmp_str, "%.2f", dns_result->latency);
            cJSON_AddStringToObject(obj, "delay", tmp_str);
            free(dial_result);
        }
        strcpy(topic, ptr->kafkaTopic);
        cJSON_AddStringToObject(obj, "domainName", ptr->domainName);
        cJSON_AddStringToObject(obj, "businessTag", ptr->businessTag);
        cJSON_AddItemToArray(arr, obj);
    }
    cJSON_AddItemToObject(result, "dnsList", arr);
    post_data = cJSON_Print(result);
    log_debug("DNS post_data:%s", post_data);
    kafka_send(post_data, strlen(post_data), dns->kafka_url, topic);
    free(post_data);
    cJSON_Delete(result);
    // pthread_cond_signal(&dns->cond);
    // pthread_mutex_unlock(&dns->mutex);
    return NULL;
}